package kafka

import (
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/spf13/viper"
)

/*
mac 启动 kafka

https://kafka.apache.org/quickstart

*/

var (
	client sarama.SyncProducer

	msgChan chan *sarama.ProducerMessage
)

func Init(addrs []string) (err error) {
	config := sarama.NewConfig()

	config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认
	config.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出⼀个partition
	config.Producer.Return.Successes = true                   // 成功交付的消息将在success channel返回

	// 连接kafka
	client, err = sarama.NewSyncProducer(addrs, config)
	if err != nil {
		fmt.Println("producer closed, err:", err)
		return
	}
	// 初始化 channel
	chanSize := viper.GetInt("collect.chan_size")
	msgChan = make(chan *sarama.ProducerMessage, chanSize)
	// 后台起一个 goroutine。 作用：将数据发送至 kafka mq
	go sendMsg()
	return
}

func sendMsg() {
	for {
		select {
		case msg := <-msgChan:
			_, _, err := client.SendMessage(msg)
			if err != nil {
				fmt.Printf("发送信息失败，错误值： %v \n", err)
				return
			}
			println("消息发送成功！")
			fmt.Println(msg.Value)
		default:
			//time.Sleep(2000 * time.Millisecond)
			//
			////zap.L().Debug("暂时没有发现新的消息出现！！！！")
			//println("暂时没有新的消息出现！！！！", time.Now().String())

		}
	}
}

func SendToKafka(topic, data string) {
	msg := &sarama.ProducerMessage{}
	msg.Topic = topic
	// 写入的是 json格式数据的字符串
	msg.Value = sarama.StringEncoder(data)

	msgChan <- msg

}
